package com.shujia.gma.ads

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo2SkuINdex {
  def main(args: Array[String]): Unit = {

    /**
      *
      * -- ---------------------------------
      * -- 使用 DDL创建MySQL中的ADS层表
      * -- 指标：1.每天每个商品对应的订单个数
      * --      2.每天每个商品对应的订单金额
      * --      3.每天每个商品对应的数量
      * -- ---------------------------------
      *
      */

    val fsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //计划器
      .inStreamingMode()
      .build()

    val fsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    fsEnv.setParallelism(1)
    val fsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(fsEnv, fsSettings)


    fsTableEnv.executeSql(
      """
        |
        |
        |-- ---------------------------------
        |-- DWD层,支付订单明细表dwd_paid_order_detail
        |-- ---------------------------------
        |CREATE TABLE dwd_paid_order_detail
        |(
        |  detail_id BIGINT,
        |  order_id BIGINT,
        |  user_id BIGINT,
        |  province_id INT,
        |  sku_id BIGINT,
        |  sku_name STRING,
        |  sku_num INT,
        |  order_price DECIMAL(10,0),
        |  create_time TIMESTAMP(0),
        |  pay_time TIMESTAMP(0)
        | ) WITH (
        |    'connector' = 'kafka',
        |    'topic' = 'dwd.paid_order_detail',
        |    'scan.startup.mode' = 'earliest-offset',
        |    'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
        |    'format' = 'changelog-json'
        |)
        |
      """.stripMargin)

    fsTableEnv.executeSql(
      """
        |
        |-- ---------------------------------
        |-- DIM(Hbase),商品维表
        |-- ---------------------------------
        |
        |CREATE TABLE dim_sku_info (
        | id BIGINT,
        | i ROW<spu_id bigint,price decimal(10,0),sku_name STRING,sku_desc STRING,weight decimal(10,2),tm_name string,base_category3_name STRING,base_category2_name STRING ,base_category1_name STRING,sku_default_img string,create_time TIMESTAMP(3)>,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'dim:sku_info',
        | 'zookeeper.quorum' = 'master:2181,node1,2181,node2:2181',
        | 'sink.buffer-flush.max-rows'='1'
        |)
        |
      """.stripMargin)

    /**
      *
      * CREATE TABLE ads_sku_index
      * (
      * sku_id BIGINT(10),
      * sku_name VARCHAR(100),
      * weight DOUBLE,
      * tm_name VARCHAR(100),
      * price DOUBLE,
      * spu_id BIGINT(10),
      * c3_name VARCHAR(100) ,
      * c2_name VARCHAR(100),
      * c1_name VARCHAR(100),
      * order_amount DOUBLE,
      * order_count BIGINT(10),
      * sku_count BIGINT(10),
      * dt varchar(100),
      * PRIMARY KEY (sku_id,dt)
      * );
      *
      */

    fsTableEnv.executeSql(
      """
        |
        |-- ---------------------------------
        |-- 使用 DDL创建MySQL中的ADS层表
        |-- 指标：1.每天每个商品对应的订单个数
        |--      2.每天每个商品对应的订单金额
        |--      3.每天每个商品对应的数量
        |-- ---------------------------------
        |CREATE TABLE ads_sku_index
        |(
        |  sku_id BIGINT,
        |  sku_name STRING,
        |  weight DOUBLE,
        |  tm_name STRING ,
        |  price DOUBLE,
        |  spu_id BIGINT,
        |  c3_name STRING ,
        |  c2_name STRING,
        |  c1_name STRING,
        |  order_amount DOUBLE,
        |  order_count BIGINT,
        |  sku_count BIGINT,
        |  dt STRING,
        |  PRIMARY KEY (sku_id,dt) NOT ENFORCED
        |) WITH (
        |    'connector' = 'jdbc',
        |    'url' = 'jdbc:mysql://master:3306/ads?useUnicode=true&characterEncoding=UTF-8',
        |    'table-name' = 'ads_sku_index',
        |    'driver' = 'com.mysql.jdbc.Driver',
        |    'username' = 'root',
        |    'password' = '123456'
        |)
        |
        |
      """.stripMargin)


    fsTableEnv.executeSql(
      """
        |
        |insert into ads_sku_index
        |select
        |a.sku_id as sku_id,
        |b.i.sku_name as sku_name,
        |b.i.weight as weight,
        |b.i.tm_name as tm_name,
        |b.i.price as price,
        |b.i.spu_id as spu_id,
        |b.i.base_category1_name as c1_name,
        |b.i.base_category2_name as c2_name,
        |b.i.base_category3_name as c3_name,
        |a.order_amount as order_amount,
        |a.order_count as order_count,
        |a.sku_count as sku_count,
        |a.dt as dt
        |from
        |(SELECT
        |      sku_id,
        |      count(distinct order_id) order_count,-- 订单数
        |      sum(order_price * sku_num) order_amount, -- 订单金额
        |      sum(sku_num) sku_count,-- 商品数量
        |      DATE_FORMAT(pay_time,'yyyy-MM-dd') dt,
        |      PROCTIME() as  ts  -- 处理时间
        |FROM dwd_paid_order_detail
        |GROUP BY sku_id,DATE_FORMAT(pay_time,'yyyy-MM-dd')) as a
        |join dim_sku_info FOR SYSTEM_TIME AS OF a.ts as b
        |on a.sku_id=b.id
        |
        |
      """.stripMargin)


  }

}
